package com.surfilter.massdata.es.core.service.impl;

import java.io.IOException;
import java.util.List;

import org.apache.log4j.Logger;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.WriteConsistencyLevel;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;

import com.surfilter.massdata.es.core.client.EsClient;
import com.surfilter.massdata.es.core.client.pool.EsClientPool;
import com.surfilter.massdata.es.core.entity.EsEntity;
import com.surfilter.massdata.es.core.entity.EsUpdateEntity;
import com.surfilter.massdata.es.core.service.EsIndexService;


public class EsIndexServiceImpl<T extends Object> extends AbstractEsService implements EsIndexService {
	
	private EsEntity<Object> esEntity = null;
	
	private EsClientPool pool = null;

	public EsIndexServiceImpl(EsEntity<Object> esEntity, EsClientPool pool) {
		super();
		this.esEntity = esEntity;
		this.pool = pool;
	}

	protected final Logger logger = Logger.getLogger(getClass());
	
	
	public void SingleIndex(String indexName,String typeName,Object entity)
			throws IOException {
		
		EsClient esClient = pool.getEsClient();
		try {
			Client client = esClient.getClient();
			client.prepareIndex(indexName, typeName).setSource(esEntity.buildEntity(entity)).execute().actionGet();
			
			client.admin().indices().prepareRefresh().execute().actionGet();
			
		} catch (Exception e) {
			logger.error("elasticsearch single insert error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
	}
	
	public void bulkIndex(String indexName,String typeName,List<?> lstEntity , int bulkSize)
			throws IOException {
		if(lstEntity == null || lstEntity.size() < 1){
			return;
		}
		
		EsClient esClient = pool.getEsClient();
		
		try {
			BulkRequestBuilder bulkRequest = esClient.getBulkRequestBuilder();
			int percount = 0;
			long start = System.currentTimeMillis();
			
				for (int i = 0,size = lstEntity.size(); i < size; i++) {
					Object entity = lstEntity.get(i);
					XContentBuilder builder = esEntity.buildEntity(entity);
							
					IndexRequestBuilder indexRequestBuilder = esClient.getIndexRequestBuilder();
					indexRequestBuilder.setIndex(indexName);
					indexRequestBuilder.setType(typeName);
//					indexRequestBuilder.setId(idStr);
					indexRequestBuilder.setSource(builder);
					//indexRequestBuilder.setReplicationType(ReplicationType.ASYNC);
					indexRequestBuilder.setConsistencyLevel(WriteConsistencyLevel.DEFAULT);
					bulkRequest.add(indexRequestBuilder);

					percount++;
					if( percount >= bulkSize ){
						insertEs(bulkRequest, start);
						percount = 0;
						start = System.currentTimeMillis();
					}
				}
				if (percount > 0 ) {
					insertEs(bulkRequest, start);
				}
//			client.admin().indices().prepareRefresh().execute().actionGet();
		} catch (Exception e) {
			logger.error("elasticsearch bulk insert error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
	}
	
	
	
	private void insertEs(BulkRequestBuilder bulkRequest, long start){
		List<ActionRequest> actionRequestList = null;
		try {
			BulkResponse response = bulkRequest
					//.setReplicationType(ReplicationType.ASYNC)
					.setConsistencyLevel(WriteConsistencyLevel.DEFAULT)
					.execute()
					.actionGet();
			if (response.hasFailures()) {
				logger.warn("索引创建失败:"+ response.buildFailureMessage());
			}
	
			actionRequestList = bulkRequest.request().requests();
			
			StringBuilder sb = new StringBuilder(256);
			sb.append("Thread:").append(Thread.currentThread().getName());
			sb.append("\t").append("").append("总索引\t").append(actionRequestList.size());
			sb.append("\t批量创建索引用时:").append(System.currentTimeMillis() - start);
			
			logger.info(sb.toString());
		} catch (Exception e) {
			logger.error("插入失败", e);
		} finally {
			actionRequestList = bulkRequest.request().requests();
			actionRequestList.clear();
		}
	}

	@Override
	public void SingleUpdate(String indexName, String typeName,
			EsUpdateEntity entity) throws IOException {

		EsClient esClient = pool.getEsClient();
		try {
			Client client = esClient.getClient();
			client.prepareUpdate().setIndex(indexName).setType(typeName)
				.setId(entity.getEsId()).setDoc(
					esEntity.buildUpdateEntity(entity))
					.execute().actionGet();
			
			esClient.getClient().admin().indices().prepareRefresh().execute().actionGet();
		} catch (Exception e) {
			logger.error("elasticsearch single insert error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
		
	}

	@Override
	public void bulkUpdate(String indexName, String typeName,
			List<? extends EsUpdateEntity> lstEntity, int bulkSize) throws IOException {
		if(lstEntity == null || lstEntity.size() < 1){
			return;
		}
		
		EsClient esClient = pool.getEsClient();
		
		try {
			BulkRequestBuilder bulkRequest = esClient.getBulkRequestBuilder();
			int percount = 0;
			long start = System.currentTimeMillis();
			
				for (int i = 0,size = lstEntity.size(); i < size; i++) {
					EsUpdateEntity entity = lstEntity.get(i);
					XContentBuilder builder = esEntity.buildUpdateEntity(entity);
							
					UpdateRequestBuilder updateRequestBuilder = esClient.getClient().prepareUpdate();
					updateRequestBuilder.setIndex(indexName);
					updateRequestBuilder.setType(typeName);
					updateRequestBuilder.setId(entity.getEsId());
					percount++;
					
					try {
						updateRequestBuilder.setDoc(builder);
					} catch (Exception e) {
						logger.error("update error", e);
						continue;
					}
					updateRequestBuilder.setConsistencyLevel(WriteConsistencyLevel.DEFAULT);
					//.setReplicationType(ReplicationType.ASYNC);
					bulkRequest.add(updateRequestBuilder);

					
					if( percount >= bulkSize ){
						insertEs(bulkRequest, start);
						percount = 0;
						start = System.currentTimeMillis();
					}
				}
				if (percount > 0 ) {
					insertEs(bulkRequest, start);
				}
			esClient.getClient().admin().indices().prepareRefresh().execute().actionGet();
		} catch (Exception e) {
			logger.error("elasticsearch single insert error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
	}
	
}
